package com.shujia;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

public class Code04GetTable {
    Connection connection;
    @Before
    public void getConnection(){
        Configuration configuration = new Configuration();

        // 注意：需要在当前平台中配置node1 node2的主机映射
        configuration.set("hbase.zookeeper.quorum","192.168.163.100,node1,node2");
        try {
            connection  = ConnectionFactory.createConnection(configuration);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    @Test
    public void getData() throws IOException {

        Table table = connection.getTable(TableName.valueOf("api:stu"));

        Get get = new Get("1500101000".getBytes());
        // Result中应该有多个Value值，rowKey对应多个列，每一列对应一个Value值
        Result result = table.get(get);

        //  public byte[] getValue(byte [] family, byte [] qualifier) {
        String name = new String(result.getValue("info".getBytes(), "name".getBytes()));
        String age = Bytes.toString(result.getValue("info".getBytes(), "age".getBytes()));
        String gender = Bytes.toString(result.getValue("info".getBytes(), "gender".getBytes()));
        String clazz = Bytes.toString(result.getValue("info".getBytes(), "clazz".getBytes()));
        System.out.println("name:"+name+"\tage:"+age+"\tgender:"+gender+"\tclazz:"+clazz);
    }

    @Test
    public void getData2() throws IOException {
        Table table = connection.getTable(TableName.valueOf("api:stu"));

        ArrayList<Get> gets = new ArrayList<>();

        gets.add(new Get("1500100999".getBytes()));
        gets.add(new Get(Bytes.toBytes("1500100998")));

        Result[] results = table.get(gets);
        for (Result result : results) {
            // 每一个单元格的数据
            List<Cell> cells = result.listCells();
            for (Cell cell : cells) {
//                cell.getValueArray()
                // 通过CellUtil工具类对cell对象中的数据进行取值
                String family = Bytes.toString(CellUtil.cloneFamily(cell));
                String colum = Bytes.toString(CellUtil.cloneQualifier(cell));
                String value = Bytes.toString(CellUtil.cloneValue(cell));
//                System.out.println("family:"+family+"\tcol:"+colum+"\tvalue:"+value);
                // 对于添加的数据是按照 rowKey + 列名称 + 时间戳 进行排序的
                System.out.print(value+"\t");
            }
            System.out.println();
        }

    }


    @After
    public void close(){
        try {
            connection.close();
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }
}
